/**
 *    Copyright (c) [2022] [中和农信]
 *    rabbit-mq-plus is licensed under Mulan PSL v2.
 *    You can use this software according to the terms and conditions of the Mulan PSL v2.
 *    You may obtain a copy of Mulan PSL v2 at:
 *             http://license.coscl.org.cn/MulanPSL2
 *    THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND, EITHER EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT, MERCHANTABILITY OR FIT FOR A PARTICULAR PURPOSE.
 *    See the Mulan PSL v2 for more details.
 *
 */
package com.cfpamf.athean.rabbit.mq.plus.consumer.core.impl;

import java.io.IOException;

import com.cfpamf.athean.rabbit.mq.plus.common.constant.BusinessConstants;
import com.cfpamf.athean.rabbit.mq.plus.domain.dto.RabbitMqPlusConsumerDTO;
import com.cfpamf.athean.rabbit.mq.plus.service.RabbitMqPlusConsumerTaskService;
import com.cfpamf.athena.utils.DateUtils;
import com.rabbitmq.client.AMQP.BasicProperties;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.DefaultConsumer;
import com.rabbitmq.client.Envelope;

import lombok.extern.slf4j.Slf4j;

/**
 * 类RabbitMqPlusBusinessMessageConsumeListener.java的实现描述：
 *
 * @author Andy（陈洪桥）2022年11月27 17:04
 */
@Slf4j
public class RabbitMqPlusBusinessMessageConsumeListener extends DefaultConsumer {

    private final RabbitMqPlusConsumerHandler rabbitMqPlusConsumerHandler;
    private final RabbitMqPlusConsumerTaskService rabbitMqPlusConsumerTaskService;
    private final RabbitMqPlusConsumerDTO consumerDTO;

    public RabbitMqPlusBusinessMessageConsumeListener(final RabbitMqPlusConsumerHandler rabbitMqPlusConsumerHandler,
        RabbitMqPlusConsumerTaskService rabbitMqPlusConsumerTaskService, final RabbitMqPlusConsumerDTO consumerDTO,
        Channel channel) {
        super(channel);
        this.rabbitMqPlusConsumerHandler = rabbitMqPlusConsumerHandler;
        this.rabbitMqPlusConsumerTaskService = rabbitMqPlusConsumerTaskService;
        this.consumerDTO = consumerDTO;
    }

    @Override
    public void handleDelivery(String consumerTag, Envelope envelope, BasicProperties properties, byte[] body)
        throws IOException {
        try {
            String messageId = properties.getMessageId();
            String data = new String(body, "UTF-8");
            // 业务消息消费失败，开始首次消费处理
            rabbitMqPlusConsumerHandler.consumerProcessor(this.rabbitMqPlusConsumerTaskService, this.consumerDTO,
                consumerTag, BusinessConstants.ZERO, DateUtils.currentTime(), messageId, data);
            // ack确认
            super.getChannel().basicAck(envelope.getDeliveryTag(), true);
        } catch (Exception e) {
            if (log.isInfoEnabled()) {
                log.info("[rabbitMqPlus][firstConsumeListener]messageId:{} queueName:{} 异常错误信息：{} 这条消息重回放入队列，重新处理",
                    properties.getMessageId(), consumerDTO.getQueueName(), e);
            }
            // 异常拒绝释放消息，设置为true表示开启重回队列 将这条消息重回放入队列
            super.getChannel().basicNack(envelope.getDeliveryTag(), false, true);
            throw e;
        }

    }

}
